package com.tobby.demo.infinispan.weather;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.infinispan.Cache;
import org.infinispan.CacheStream;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.stream.CacheCollectors;

public class WeatherApp {

	// static String[] locations = { "Rome, Italy", "Como, Italy", "Basel, Switzerland",
	// "Bern, Switzerland", "London, UK", "Newcastle, UK", "Bucarest, Romania",
	// "Cluj-Napoca, Romania", "Ottawa, Canada", "Toronto, Canada",
	// "Lisbon, Portugal", "Porto, Portugal", "Raleigh, USA", "Washington, USA" };

	static String[] locations = { "Rome, Italy", "Como, Italy", "Basel, Switzerland",
			"Bern, Switzerland", "Bucarest, Romania", "Cluj-Napoca, Romania",
			"London, UK", "Newcastle, UK" };

	private final EmbeddedCacheManager cacheManager;

	private final WeatherService weatherService;

	private Cache<String, LocationWeather> cache;

	private final ClusterListener listener;

	public WeatherApp() throws InterruptedException, IOException {
		// GlobalConfigurationBuilder global = GlobalConfigurationBuilder
		// .defaultClusteredBuilder();
		// global.transport().clusterName("WeatherAppTobby");
		// ConfigurationBuilder config = new ConfigurationBuilder();
		// config.expiration().lifespan(5, TimeUnit.SECONDS).clustering()
		// .cacheMode(CacheMode.DIST_SYNC).hash().groups().enabled()
		// .addGrouper(new LocationWeather.LocationGroup());
		// cacheManager = new DefaultCacheManager(global.build(), config.build());
		cacheManager = new DefaultCacheManager("weatherapp-infinispan.xml");
		listener = new ClusterListener(2);
		cacheManager.addListener(listener);
		cache = cacheManager.getCache();
		cache.addListener(new CacheListener());
		weatherService = initWeatherService(cache);

		System.out.println("-----Wait for cluster to form -----");
		listener.clusterFormedLatch.await();
	}

	private WeatherService initWeatherService(Cache<String, LocationWeather> cache) {
		String apiKey = System.getenv("OWMAPIKEY");
		// apiKey = "03409758c3bd8382744ca6043a86daf8";
		if (apiKey == null) {
			System.out
					.println("WARNING: OWMAPIKEY environment variable not set, using the RandomWeatherService.");
			return new RandomWeatherService(cache);
		}
		else {
			return new OpenWeatherMapService(apiKey, cache);
		}
	}

	public void fetchWeather() {
		System.out.println("---Fetching Weather Infomation---");
		long start = System.currentTimeMillis();
		List<String> locationList = Arrays.asList(locations);
		locationList.forEach((location) -> {
			LocationWeather weather = weatherService.getWeatherForLocation(location);
			System.out.printf("%s - %s\n", location, weather);
		});
		System.out.printf("---Fetched in %dms ---\n", System.currentTimeMillis() - start);
	}

	public void computeCountryAverages() {
		System.out.println("---- Average country temperatures ----");
		// Map<String, Double> averages = ((Stream<Entry<String, LocationWeather>>) cache
		// .entrySet().stream()).collect(Collectors.groupingBy(
		// e -> e.getValue().country, Collectors.averagingDouble((
		// Entry<String, LocationWeather> e) -> e.getValue().temperature)));
		Map<String, Double> averages = ((Stream<Entry<String, LocationWeather>>) cache
				.entrySet().stream()).collect(CacheCollectors
				.serializableCollector(() -> Collectors.groupingBy(
						e -> e.getValue().country,
						Collectors
								.averagingDouble((Entry<String, LocationWeather> e) -> e
										.getValue().temperature))));
		for (Entry<String, Double> entry : averages.entrySet()) {
			System.out.printf("Average temperature in %s is %.1f° C\n", entry.getKey(),
					entry.getValue());
		}
	}

	public void shutdown() throws InterruptedException {
		if (!cacheManager.isCoordinator()) {
			listener.shutdownLatch.await();
		}
		cacheManager.stop();
	}

	public static void main(String[] args) throws InterruptedException, IOException {
		WeatherApp app = new WeatherApp();

		if (app.cacheManager.isCoordinator()) {
			app.fetchWeather();
			app.computeCountryAverages();
			TimeUnit.SECONDS.sleep(5);
			app.fetchWeather();
		}
		app.shutdown();
	}

}
